package com.dashu.device.util;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDB.ConsistencyLevel;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Point.Builder;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

/**
 * InfluxDB数据库连接操作类
 *
 * @author 大脑补丁
 */
public class InfluxDBConnection {

    private static InfluxDB influxDB;
    // 用户名
    private String username = "admin";
    // 密码
    private String password = "admin";
    // 连接地址
    private String openurl = "http://192.168.101.28:8086";
    // 数据库
    private static String database = "dashu";
    // 保留策略
    private static String retentionPolicy;


    public InfluxDBConnection(String retentionPolicy) {
        this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;
        influxDbBuild();
    }

    public static InfluxDB  getInfluxDB() {
        InfluxDBConnection influxDBConnection = new InfluxDBConnection( "");
        return influxDB;
    }

    /**
     * 创建数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public static void createDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /**
     * 删除数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public static void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public static boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 连接时序数据库 ，若不存在则创建
     *
     * @return
     */
    public  InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(openurl, username, password);
        }
        try {
            // if (!influxDB.databaseExists(database)) {
            // influxDB.createDatabase(database);
            // }
        } catch (Exception e) {
            // 该数据库可能设置动态代理，不支持创建数据库
            // e.printStackTrace();
        } finally {
            influxDB.setRetentionPolicy(retentionPolicy);
        }
        influxDB.setLogLevel(InfluxDB.LogLevel.NONE);
        return influxDB;
    }

    /**
     * 创建自定义保留策略
     *
     * @param policyName  策略名
     * @param duration    保存天数
     * @param replication 保存副本数量
     * @param isDefault   是否设为默认保留策略
     */
    public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
                database, duration, replication);
        if (isDefault) {
            sql = sql + " DEFAULT";
        }
        this.query(sql);
    }

    /**
     * 创建默认的保留策略
     *
     * @param //策略名：default，保存天数：30天，保存副本数量：1 设为默认保留策略
     */
    public  void createDefaultRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "default", database, "30d", 1);
        this.query(command);
    }

    /**
     * 查询
     *
     * @param command 查询语句
     * @return
     */
    public static  QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }

    /**
     * 插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public  static void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time,
                       TimeUnit timeUnit) {
        Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);
        if (0 != time) {
            builder.time(time, timeUnit);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public  static void batchInsert(BatchPoints batchPoints) {
        influxDB.write(batchPoints);
        // influxDB.enableGzip();
        // influxDB.enableBatch(2000,100,TimeUnit.MILLISECONDS);
        // influxDB.disableGzip();
        // influxDB.disableBatch();
    }

    /**
     * 批量写入数据
     *
     * @param database        数据库
     * @param retentionPolicy 保存策略
     * @param consistency     一致性
     * @param records         要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public static void batchInsert(final String database, final String retentionPolicy, final ConsistencyLevel consistency,
                            final List<String> records) {
        influxDB.write(database, retentionPolicy, consistency, records);
    }

    /**
     * 删除
     *
     * @param command 删除语句
     * @return 返回错误信息
     */
    public static  String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 关闭数据库
     */
    public  void close() {
        influxDB.close();
    }

    /**
     * 构建Point
     *
     * @param measurement
     * @param time
     * @param fields
     * @return
     */
    public static Point pointBuilder(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {
        Point point = Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
        return point;
    }

    public static void main(String[] args) {
        getInfluxDB();
        createDB("test");
//        deleteDB("test");
        List<String> databases = influxDB.describeDatabases();
        System.out.println("数据库的数量为：{},列表如下：{}" + databases.size() + databases);
//        //插入用的tag、field两个Map集合
//        Map<String,String> tags = new HashMap<String, String>();
//        Map<String,Object> fields = new HashMap<String, Object>();
//        tags.put("name","lisi");
//        tags.put("sex","nv");
//        tags.put("id","0001");
//        fields.put("english",85);
//        fields.put("math",81); b
        //调用插入方法
//        influxJDBC.insert("student",tags,fields);
//        influxDBConnection.insert("weather", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
//        QueryResult query =     influxDBConnection.query( "select * from weather where math = 81");
//        System.out.println(query);
//        countSeries();
    }


    public static void countSeries() {
        Query query = null;
        QueryResult queryResult = null;
        String sql = null;

        List<String> databases = influxDB.describeDatabases();
        databases.remove("_internal");
        System.out.println("数据库的数量为：{},列表如下：{}" + databases.size() + databases);
//        LOGGER.info("数据库的数量为：{},列表如下：{}", databases.size(), databases);

        for (String databaseName : databases) {
            influxDB.setDatabase(databaseName);

            sql = "show measurements";
            query = new Query(sql, databaseName);
            queryResult = influxDB.query(query);

            List<String> measurements = getMeasurements(queryResult);
            for (String measurementName : measurements) {
                sql = String.format("SHOW SERIES ON \"%s\" FROM \"%s\"", databaseName, measurementName);
                query = new Query(sql, databaseName);
                queryResult = influxDB.query(query);
                long num = queryResult.getResults().get(0).getSeries().get(0).getValues().size();
                System.out.println("{}.{}.seriesCount = {}" + databaseName + measurementName + num);
//                LOGGER.info("{}.{}.seriesCount = {}", databaseName, measurementName, num);
            }
        }

    }

    private static List<String> getMeasurements(QueryResult queryResult) {
        List<String> collect = queryResult.getResults().get(0).getSeries().get(0).getValues().stream()
                .map(x -> (String) x.get(0)).collect(Collectors.toList());
        return collect;
    }


}


